package com.codejiwei.test

import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import scala.util.Random

/**
 * Author: jiwei01
 * Date: 2022/8/22 10:38
 * Package: com.codejiwei.test
 * Description:
 */
object StreamSample{
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(2)
    env.addSource(MySourceFunction()).print()

    env.execute("TestFlink1.14.5WithStreamingJar")
  }
}

case class MySourceFunction() extends SourceFunction[String] {
  private var isRunning = true
  override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
    val random = new Random()
    while (isRunning) {
      ctx.collect(random.toString)
      Thread.sleep(2000)
    }
  }

  override def cancel(): Unit = {
    isRunning = false
  }
}
